package com.sjc.lesson04.demo05;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * IN, OUT, KEY, W
 * IN: 输入的数据类型
 * OUT: 输出的数据类型
 * KEY: key的数据类型（在Flink里面，String用Tuple表示）
 * W: window的数据类型
 */
public class SumProcessWindowFunction extends ProcessWindowFunction<Tuple2<String,Long>, Tuple2<String,Integer>, Tuple, TimeWindow> {

    /**
     * 当一个window触发计算时会调用这个方法
     * @param tuple         key
     * @param context       operator的上下文
     * @param elements      指定window的所有元素
     * @param collector     用户输出
     * @throws Exception
     */
    @Override
    public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements,
                        Collector<Tuple2<String, Integer>> collector) throws Exception {

        int sum = 0;
        for(Tuple2<String,Long> ele : elements){
            sum +=1;
        }
        // 输出单词出现的次数
        collector.collect(Tuple2.of(tuple.getField(0),sum));
    }
}
